package com.fzk.ha;

import com.alibaba.fastjson2.JSON;
import com.fzk.core.FileMeta;
import com.fzk.log.Logger;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.FileRegion;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.nio.charset.StandardCharsets;

public abstract class HAMsg {
    public static final byte Ping = 1;// 心跳
    public static final byte Pong = 2;// 心跳
    public static final byte Sync_Req = 3;// 文件同步请求
    public static final byte Sync_Res = 4;// 文件同步响应
    // public static final byte raft = 3;

    public int len;// 消息长度: 用于netty的TCP拆包解包
    public byte type;// 消息类型
    public static final int TypePos = 4;

    public static byte getMsgType(ByteBuf buf) {
        return buf.getByte(TypePos);
    }

    public HAMsg(int len, byte type) {
        this.len = len;
        this.type = type;
    }

    // 心跳消息, ping/pong
    public static class HAHeartMsg extends HAMsg {
        // 应该需要放一些连接信息，比如连接上自己的replica和自己连上的replica
        public byte[] content;

        public HAHeartMsg(byte type, byte[] content) {
            super(1 + content.length, type);
            this.content = content;
        }

        public HAHeartMsg(ByteBuf buf) {
            super(buf.readInt(), buf.readByte());
            int contentLen = this.len - 1;// 计算内容长度
            if (contentLen != buf.readableBytes()) {
                String str = String.format("heartbeat msg, 计算的contentLen: %d 不等于 buf的剩余可读字节数: %d ?", contentLen, buf.readableBytes());
                Logger.error(str);
                throw new RuntimeException(str);
            }
            byte[] dest = new byte[contentLen];
            buf.readBytes(dest, 0, contentLen);
            this.content = dest;
        }

        public ByteBuf serialize() {
            ByteBuf buf = ByteBufAllocator.DEFAULT.directBuffer(4 + this.len);
            buf.writeInt(this.len);
            buf.writeByte(this.type);
            buf.writeBytes(this.content);
            return buf;
        }

        @Override
        public String toString() {
            return "HAHeartMsg{" +
                    "type=" + type +
                    ", len=" + len +
                    ", content=" + new String(content, StandardCharsets.UTF_8) +
                    '}';
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class HASyncRequestHeader {
        public static final byte NormalSync = 0;// 0.正常持续同步
        public static final byte SingleFileSync = 1;// 1.单个文件同步
        private byte haSyncType;// 0.正常持续同步 1.单个文件同步

        private long offset;// 文件元信息存储位置
        private long position;// 请求文件起点
    }

    public static class HASyncRequestMsg extends HAMsg {
        public final int headerLen;// 请求头长度
        public byte[] headerBytes;
        public HASyncRequestHeader header;// 请求头

        public HASyncRequestMsg(HASyncRequestHeader header) {
            this(header, JSON.toJSONString(header).getBytes(StandardCharsets.UTF_8));
        }

        public HASyncRequestMsg(HASyncRequestHeader header, byte[] headerBytes) {
            super(1 + 4 + headerBytes.length, Sync_Req);
            this.headerLen = headerBytes.length;
            this.headerBytes = headerBytes;
            this.header = header;
        }

        public static HASyncRequestMsg deserialize(ByteBuf buf) {
            int len = buf.readInt();
            if (len != buf.readableBytes()) {
                throw new RuntimeException(String.format("HA sync req, len: %d和buf可读字节数: %d不一致?", len, buf.readableBytes()));
            }
            byte type = buf.readByte();
            if (type != Sync_Req) {
                throw new RuntimeException(String.format("HA sync req, type: %d不是HAMsg.Sync_Req: %d类型", type, Sync_Req));
            }
            int headerLen = buf.readInt();
            if (headerLen < 0) {
                throw new RuntimeException(String.format("HA sync req, headerLen: %d 怎么会小于0?", headerLen));
            }
            if (headerLen != buf.readableBytes()) {
                throw new RuntimeException(String.format("HA sync req, headerLen: %d 怎么会不等于buf的剩余可读字节数: %d?", headerLen, buf.readableBytes()));
            }
            if (headerLen > 0) {
                byte[] headerBytes = new byte[headerLen];
                buf.readBytes(headerBytes, 0, headerLen);
                HASyncRequestHeader header = JSON.parseObject(new String(headerBytes, StandardCharsets.UTF_8), HASyncRequestHeader.class);
                return new HASyncRequestMsg(header, headerBytes);
            }
            // 一般不会无头部
            return new HASyncRequestMsg(new HASyncRequestHeader(), new byte[0]);
        }

        // 注意需要合理释放分配的直接内存ByteBuf
        public ByteBuf serialize() {
            ByteBuf buf = PooledByteBufAllocator.DEFAULT.directBuffer(4 + 1 + 4 + this.headerLen, 4 + 1 + 4 + this.headerLen);
            buf.writeInt(this.len);
            buf.writeByte(this.type);
            buf.writeInt(this.headerLen);
            buf.writeBytes(this.headerBytes);
            return buf;
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class HASyncResponseHeader {
        private byte haSyncType;// 0.正常持续同步 1.单个文件同步

        private long offset;// 文件元信息存储位置
        private long position;// 此消息携带的文件内容起始点

        // ---元信息---
        private int metaLen;
        private FileMeta meta;
        // ---元信息---
    }

    public static class HASyncResponseMsg extends HAMsg {
        // 空响应，用于主节点无文件内容时发送给从节点，从节点需要判断是否为空响应
        public static final HASyncResponseMsg Res_Empty = new HASyncResponseMsg(null, new byte[0], (FileRegion) null);

        public final int headerLen;
        public byte[] headerBytes;
        public HASyncResponseHeader header;
        // 一般情况下，下面这两个只有1个有值
        public ByteBuf content;// 文件内容，从节点收到主节点发来的内容时有值
        public FileRegion fileRegion;// 文件块，主节点发送文件内容是有值，可用FileRegion#transferTo()减少内存拷贝

        public HASyncResponseMsg(HASyncResponseHeader header, FileRegion fileRegion) {
            this(header, JSON.toJSONString(header).getBytes(StandardCharsets.UTF_8), fileRegion);
        }

        public HASyncResponseMsg(HASyncResponseHeader header, byte[] headerBytes, FileRegion fileRegion) {
            super(1 + 4 + headerBytes.length + (int) (fileRegion == null ? 0 : fileRegion.count()), Sync_Res);
            this.headerLen = headerBytes.length;
            this.header = header;
            this.headerBytes = headerBytes;
            this.fileRegion = fileRegion;
        }

        public HASyncResponseMsg(HASyncResponseHeader header, ByteBuf content) {
            this(header, JSON.toJSONString(header).getBytes(StandardCharsets.UTF_8), content);
        }

        public HASyncResponseMsg(HASyncResponseHeader header, byte[] headerBytes, ByteBuf content) {
            super(1 + 4 + headerBytes.length + content.readableBytes(), Sync_Res);
            this.headerLen = headerBytes.length;
            this.header = header;
            this.headerBytes = headerBytes;
            this.content = content;
        }

        /**
         * 反序列化
         * 在metaLen为0时直接返回特殊值Res_Empty
         *
         * @param buf 消息二进制ByteBuf
         * @return 解析后消息实体
         */
        public static HASyncResponseMsg deserialize(ByteBuf buf) {
            int len = buf.readInt();
            if (len != buf.readableBytes()) {
                throw new RuntimeException(String.format("HA sync res, len: %d和buf可读字节数: %d不一致?", len, buf.readableBytes()));
            }
            byte type = buf.readByte();
            if (type != Sync_Res) {
                throw new RuntimeException(String.format("HA sync res, type: %d不是HAMsg.Sync_Res: %d类型", type, Sync_Res));
            }
            int headerLen = buf.readInt();
            if (headerLen < 0) {
                throw new RuntimeException(String.format("headerLen: %d <0 ???", headerLen));
            }
            if (headerLen == 0) {// 说明此时主节点返回空数据
                return Res_Empty;
            }
            // 注意：直接内存ByteBuf不能切片
            byte[] headerBytes = new byte[headerLen];
            buf.readBytes(headerBytes, 0, headerLen);
            HASyncResponseHeader header = JSON.parseObject(new String(headerBytes, StandardCharsets.UTF_8), HASyncResponseHeader.class);

            // 计算文件内容长度
            int contentLen = len - 1 - 4 - headerLen;
            if (contentLen != buf.readableBytes()) {
                String str = String.format("HA sync res, 计算的contentLen: %d 不等于 buf的剩余可读字节数: %d ?", contentLen, buf.readableBytes());
                Logger.error(str);
                throw new RuntimeException(str);
            }

            // 剩余作为文件内容
            // 注意：直接内存ByteBuf不能切片
            // 注意：此ByteBuf的生命周期，即其在哪里被释放资源的
            return new HASyncResponseMsg(header, headerBytes, buf);
        }

        /**
         * 转换为ByteBuf进行消息发送
         * 注意：这里尽量不要发生内存拷贝
         * 所以采用了组合ByteBuf
         * 注意需要合理的释放直接内存ByteBuf
         *
         * @return compositeByteBuf
         */
        public ByteBuf serialize() {
            ByteBuf buf = ByteBufAllocator.DEFAULT.directBuffer(4 + 1 + 4 + this.headerLen, 4 + 1 + 4 + this.headerLen);
            buf.writeInt(this.len);
            buf.writeByte(this.type);
            buf.writeInt(this.headerLen);
            if (this.headerLen > 0) {
                buf.writeBytes(this.headerBytes);
            }

            if (content == null) {// 没有ByteBuf时直接返回即可，常用于主节点发送文件内容时
                return buf;
            }
            CompositeByteBuf compositeByteBuf = ByteBufAllocator.DEFAULT.compositeDirectBuffer();
            compositeByteBuf.addComponents(buf, this.content);
            return compositeByteBuf;
        }
    }
}
